Summary

• Rebuilds the page-stream → Arrow decoder to be page-bounded — StreamDecoder::decode_next_page() returns one [DecodedPage] per call (rg_idx, col_idx, page_idx, row_start, ArrayRef) instead of materialising an entire row group at a time.
• Memory: ~one in-flight page (compressed + decompressed bytes) + one cached dictionary page per (rg, col) when dict-encoded. The decoder does NOT buffer a row group, column chunk, or any materialised array beyond the one currently being emitted.
• PR-6b's merge engine consumes [DecodedPage]s in storage order (row-group-major, column-major-within-rg, page-major-within-col), applies merge plan slicing per page, and streams output pages directly into the writer without column-chunk staging.

How it works

1. Pull one [Page] from the underlying [ColumnPageStream]. Skip INDEX_PAGE (historical Thrift variant, not emitted by production writers).
2. Look up or initialise per-(rg, col) state: a PageQueue that feeds parquet-rs's [ColumnReader] one page at a time, plus a counter tracking rows decoded so far.
3. Convert the [Page] to parquet-rs's column::page::Page enum: decompress via [parquet::compression::create_codec] (requires the experimental feature), translate format::Encoding (Thrift wrapper) → basic::Encoding (Rust enum) via a manual i32 match (no public conversion in parquet-rs), drop optional statistics.
4. Push the converted page onto the queue. Dictionary/index pages absorb silently for use by subsequent data pages.
5. For a data page: ask the [ColumnReader] to decode exactly header.num_values records via read_records(...) calls in a loop, pulling values + def/rep levels into typed buffers.
6. Build an ArrayRef from (values, def_levels, rep_levels) per the column's parquet physical type. Emit [DecodedPage].

Type coverage

Flat physical types: Boolean, Int8/16/32 + UInt8/16/32 (parquet Int32 with logical annotation), Int64/UInt64 (parquet Int64), Float32, Float64, Utf8/LargeUtf8/Binary/LargeBinary (parquet ByteArray). Dictionary-encoded pages are decoded via the cached dict page → values pipeline.

List<T> / LargeList<T> where outer + inner are non-nullable and inner is a flat primitive — covers DDSketch keys (List<Int16>) and counts (List<UInt64>). Dremel def/rep levels (max_def=1, max_rep=1) are decoded via the same read_records path; arrow offsets are computed via list_offsets_from_levels.

Other nested shapes (nullable list inner/outer, Struct, Map, FixedSizeList, multi-leaf nested) return an unsupported-type error rather than silently falling back to a different mechanism.

Sync ⇄ async bridging

The page stream is async (S3 reads); PageReader (from parquet-rs) is sync. Bridged via Arc<Mutex<VecDeque<ColumnPage>>> per (rg, col): decode_next_page pulls from the stream (async), pushes onto the queue, then the sync PageReader impl pops from the queue when the ColumnReader asks for the next page. peek_next_page and skip_next_page are properly implemented to support the parquet-rs reader's state machine.

Schema handling

parquet_to_arrow_schema(parquet_schema, None) bypasses the ARROW:schema hint that would otherwise force Dictionary types — input parquet files written from arrow declare Dictionary columns in ARROW:schema metadata, but their page-encoded values are plain values that decode to StringArray/etc. Decoding without the hint gives consistent flat-primitive output that the merge engine then interleaves.

Tests

9 tests, all passing:

• test_drain_single_rg_round_trip, test_drain_multi_rg_round_trip — full round-trip via decode_next_page matches ParquetRecordBatchReaderBuilder.
• test_decoded_page_row_indexing — row_start correctly tracks per-(rg, col) row offsets.
• test_eof_idempotent — repeated calls after EOF stay Ok(None).
• test_nullable_column_round_trip — def-level decoding for nullable cols.
• test_compression_codecs — snap, gzip, zstd round-trip.
• test_page_bounded_queue_depth — verifies the internal page queue depth stays ≤ 2 across a long stream (the page-bounded contract).
• test_list_uint64_round_trip — List<UInt64> (DDSketch shape) round-trip.
• test_io_failure_surfaces_as_page_stream_error — body GET failures propagate as PageStream(Io), not masked as decode errors.

Stack

Base: gtt/column-page-stream-trait (PR-5a #6406).

PR-6b (#6409) builds the streaming merge engine on top of this decoder.